package com.atguigu.realtime.util;

import com.atguigu.realtime.common.Constant;

/**
 * @Author lzc
 * @Date 2023/3/13 09:52
 */
public class SQLUtil {
    
    public static String getKafkaSourceDDL(String topic, String groupId, String... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        
        return "with(" +
            "  'connector' = 'kafka'," +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "'," +
            "  'properties.group.id' = '" + groupId + "'," +
            "  'topic' = '" + topic + "'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            (defaultFormat.equals("json") ? "  'json.ignore-parse-errors' = 'true',  " : "") + // 当解析失败的是, 是否跳过这个行数据
            "  'format' = '" + defaultFormat + "'  " +
            ")";
    }
    
    public static String getKafkaSinkDDL(String topic, String... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        
        return "with(" +
            "  'connector' = 'kafka'," +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "'," +
            "  'topic' = '" + topic + "'," +
            "  'sink.semantic' = 'exactly-once'," +
            "  'properties.transaction.timeout.ms' = '900000', " +
            "  'format' = '" + defaultFormat + "'  " +
            ")";
    }
    
    public static String getUpsertKafkaDDL(String topic, String... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        
        return "with(" +
            "  'connector' = 'upsert-kafka'," +
            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "'," +
            "  'topic' = '" + topic + "'," +
            (defaultFormat.equals("json") ? "  'value.json.ignore-parse-errors' = 'true',  " : "") + // 当解析失败的是, 是否跳过这个行数据
            (defaultFormat.equals("json") ? "  'key.json.ignore-parse-errors' = 'true',  " : "") + // 当解析失败的是, 是否跳过这个行数据
            "  'key.format' = '" + defaultFormat + "'," +
            "  'value.format' = '" + defaultFormat + "' " +
            ")";
    }
}
